package com.bw.gamll.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class CommentApp extends BaseSqlApp {
    public static void main(String[] args) {
        new CommentApp().start(Constant.TOPIC_DB,Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO,4,10012);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        /**
         * 1.通过flinksql把评论数据筛选出来
         * 2.读取码表数据，数据在Hbase
         * 3.评论跟码表进行join操作
         * 4、写出到kakfka主题
         */
        readOdsDb(tableEnv,Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
////        1.通过 flinksql 把评论数据筛选出来
        Table commentTab = getCommentTab(tableEnv);
        tableEnv.createTemporaryView("comment_info",commentTab);

//        tableEnv.sqlQuery("SELECT * FROM base_dic").execute().print();

//        2.读取码表数据，数据在Hbas
        readBaseDic(tableEnv);
//        3.评论跟码表进行join操作
        // 用Lookup join需要用到处理时间,用大表和小表

        Table commentJoin = getCommentJoin(tableEnv);

//        4、写出到kakfka主题
        extracted(tableEnv);


        commentJoin.insertInto(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO).execute();
    }

    private static void extracted(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO+"(\n" +
                "    id string,\n" +
                "    user_id string,\n" +
                "    nick_name string,\n" +
                "    sku_id string,\n" +
                "    spu_id string,\n" +
                "    order_id string,\n" +
                "    appraise_code string,\n" +
                "    appraise_name string,\n" +
                "    comment_txt string,\n" +
                "    create_time string\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO));
    }


    private Table getCommentJoin(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select\n" +
                "    id,\n" +
                "    user_id,\n" +
                "    nick_name,\n" +
                "    sku_id,\n" +
                "    spu_id,\n" +
                "    order_id,\n" +
                "    appraise,\n" +
                "    dic_name appraise_name,\n" +
                "    comment_txt,\n" +
                "    create_time\n" +
                "from comment_info as c\n" +
                "         JOIN base_dic FOR SYSTEM_TIME AS OF c.proc_time AS b\n" +
                "    ON c.appraise = b.rowkey");
    }

    private static Table getCommentTab(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select\n" +
                "  `data`['id'] id,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['nick_name'] nick_name,\n" +
                "  `data`['sku_id'] sku_id,\n" +
                "  `data`['spu_id'] spu_id,\n" +
                "  `data`['order_id'] order_id,\n" +
                "  `data`['appraise'] appraise,\n" +
                "  `data`['comment_txt'] comment_txt,\n" +
                "   proc_time,\n" +
                "  `data`['create_time'] create_time\n" +
                "from topic_db\n" +
                "where `database` = 'gmall' and `table` = 'comment_info' and `type`='insert'");
    }
}
